package com.example


import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession, functions}

// IP 命中率(Hit/Miss) 响应时间 请求时间 请求方法 请求URL 请求协议 状态码 响应大小 referer 用户代理
//case class CdnInfo(ip: String, hitOrMiss: String,
//                   responseTime: Double, requestTime: String,
//                   requestMethod: String, requestURL: String,
//                   protocol: String, status: Int,
//                   response: Int, referer: String,
//                   userAgent: String
//                  )

object Test2 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName(this.getClass.getCanonicalName)
      .master("local[*]")
      .getOrCreate()
    val sc = spark.sparkContext
    sc.setLogLevel("warn")
    import spark.implicits._

//    val cdnSchema = StructType(StructField("ip", StringType, true) ::
//      StructField("hitOrMiss", StringType, true) ::
//      StructField("responseTime", DoubleType, true) ::
//      StructField("requestTime", StringType, true) ::
//      StructField("requestMethod", StringType, true) ::
//      StructField("requestURL", StringType, true) ::
//      StructField("protocol", StringType, true) ::
//      StructField("status", IntegerType, true) ::
//      StructField("response", IntegerType, true) ::
//      StructField("referer", StringType, true) ::
//      StructField("userAgent", StringType, true) :: Nil)
//    val cdnDf: DataFrame = spark.read
//      .option("delimiter", " ")
//      .option("inferschema", "true")
//      .schema(cdnSchema)
//      .csv("data/cdn.txt")

    val cdnRawRdd: RDD[String] = sc.textFile("data/cdn.txt")
    println(countIP(cdnRawRdd))
    videoIPCount(cdnRawRdd)
    hourCount(cdnRawRdd)

  }

  def countIP(cdnRDD: RDD[String]): Long = {
    val ipCounts = cdnRDD.map(line => {
      val fields: Array[String] = line.split("\\s+")
      fields(0)
    }).distinct().count()
    ipCounts
  }

  def videoIPCount(cdnRDD: RDD[String]): Unit ={
    val rdd: RDD[(String, String)] = cdnRDD.map(line => {
      val fields: Array[String] = line.split("\\s+")
      (fields(0), fields(6))
    })
    // 筛选出带.mp4的url
    val videoRdd: RDD[(String, String)] = rdd.filter(_._2.endsWith(".mp4"))
    val value: RDD[(String, String)] = videoRdd.map(x => {
      (x._2, x._1)
    })
    // 根据url为key进行分组，然后汇总得到每个视频的IP数量，最后输出文件
    val value1: RDD[(String, Iterable[String])] = value.groupByKey()
    value1.map(x => {
      (x._1, x._2.toSet.size)
    }).repartition(1).saveAsTextFile("data/output/output_videoIPCount")
  }

  def hourCount(cdnRDD: RDD[String]) = {
    val rdd: RDD[(String, String)] = cdnRDD.map(line => {
      val fields: Array[String] = line.split("\\s+")
      (fields(3).split("/")(2).split(":")(1), fields(6))
    })

    rdd.groupByKey().map(x => {
      (x._1, x._2.size)
    }).repartition(1).sortByKey().saveAsTextFile("data/output/output_hourCount")
  }
}
